package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.ExamPaperIntervalScoreBean;
import com.atguigu.edu.realtime220815.bean.ExamQuestionAnswerBean;
import com.atguigu.edu.realtime220815.bean.ExamQuestionAnswerUserBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/19 0:12
 **/
public class DwsExamQuestionAnswerWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_exam_question_detail";
        String groupId = "DwsExamQuestionAnswerWindow_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kfkDS = env.addSource(flinkKafkaConsumer);
//{"id":"495965","exam_id":"33308","paper_id":"188","question_id":"744","user_id":"2408","is_correct":"0","score":"0.0","date_id":"2023-02-18","ts":"1676703783"}
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kfkDS.map(JSON::parseObject);

        SingleOutputStreamOperator<ExamQuestionAnswerBean> beanDS = jsonObjDS.map(new MapFunction<JSONObject, ExamQuestionAnswerBean>() {
            @Override
            public ExamQuestionAnswerBean map(JSONObject value) throws Exception {
                String question_id = value.getString("question_id");
                Long ts = value.getLong("ts");
                String is_correct = value.getString("is_correct");
                Integer correct_count = 0;
                if ("1".equals(is_correct)){
                    correct_count = 1;
                }

                ExamQuestionAnswerBean examQuestionAnswerBean = ExamQuestionAnswerBean.builder()
                        .questionId(question_id)
                        .correct_count(correct_count)
                        .ts(ts)
                        .build();
                return examQuestionAnswerBean;
            }
        });

        SingleOutputStreamOperator<ExamQuestionAnswerBean> withWaterDS
                = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ExamQuestionAnswerBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<ExamQuestionAnswerBean>() {
                    @Override
                    public long extractTimestamp(ExamQuestionAnswerBean element, long recordTimestamp) {
                       return element.getTs()*1000;
                    }
                }));

        KeyedStream<ExamQuestionAnswerBean, String> keyDS = withWaterDS.keyBy(value -> value.getQuestionId());

        WindowedStream<ExamQuestionAnswerBean, String, TimeWindow> windowDS = keyDS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<ExamQuestionAnswerBean> reduceDS = windowDS.reduce(
                new ReduceFunction<ExamQuestionAnswerBean>() {
                    @Override
                    public ExamQuestionAnswerBean reduce(ExamQuestionAnswerBean value1, ExamQuestionAnswerBean value2) throws Exception {
                        value1.setAnswer_count(value1.getAnswer_count() + value2.getAnswer_count());
                        value1.setCorrect_count(value1.getCorrect_count() + value2.getAnswer_count());

                        return value1;
                    }
                },
                new WindowFunction<ExamQuestionAnswerBean, ExamQuestionAnswerBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamQuestionAnswerBean> input, Collector<ExamQuestionAnswerBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();

                        for (ExamQuestionAnswerBean examQuestionAnswerBean : input) {
                            examQuestionAnswerBean.setStt(DateFormatUtil.toYmdHms(start));
                            examQuestionAnswerBean.setEdt(DateFormatUtil.toYmdHms(end));
                            examQuestionAnswerBean.setTs(now);
                            out.collect(examQuestionAnswerBean);
                        }
                    }
                }
        );

        SingleOutputStreamOperator<ExamQuestionAnswerBean> withQuestionNameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamQuestionAnswerBean>("dim_test_question_info") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, ExamQuestionAnswerBean obj) {
                        String question_name = dimInfoJsonObj.getString("question_txt".toUpperCase());
                        obj.setQuestionName(question_name);
                    }

                    @Override

                    public String getKey(ExamQuestionAnswerBean obj) {
                        return obj.getQuestionId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        withQuestionNameDS.print();
        withQuestionNameDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_exam_question_answer_window values(?,?,?,?,?,?,?)"
        ));

        env.execute();
    }
}
